[flink] Support bounded streaming read for Fluss source - #4022
[flink] Support bounded streaming read for Fluss source#4022naivedogger wants to merge 9 commits into
Conversation
Similar to the Kafka connector's bounded read, a streaming job can now read from a given starting position up to a given stopping position and then finish, which is useful for replaying a bounded time range of the log, backfilling and archiving. - FlinkSource reports BOUNDED when stopping offsets are supplied, and passes them through createEnumerator/restoreEnumerator. - FlinkSourceEnumerator treats a streaming read with stopping offsets as bounded: one-time partition discovery and NoMoreSplits signaling, so the job finishes once all splits reach their stopping offsets. - scan.bounded.mode is supported for log tables, the changelog of primary key tables (earliest/latest/timestamp startup mode) and the $changelog/$binlog virtual tables; the full startup mode of primary key tables and the datalake union read are rejected explicitly. - FlussSourceBuilder#setBounded(OffsetsInitializer) enables bounded streaming reads in the DataStream API.
loserwang1024
left a comment
loserwang1024
left a comment
There was a problem hiding this comment.
I have left some comment. maybe we can keep the same behavior with kafka to simplify the model.
| "Only OffsetsInitializer.latest() and OffsetsInitializer.timestamp(...) are " | ||
| + "supported as stopping offsets, but was %s.", | ||
| checkedStoppingOffsetsInitializer.getClass().getName()); | ||
| this.stoppingOffsetsInitializer = checkedStoppingOffsetsInitializer; |
There was a problem hiding this comment.
please set private boolean bounded; as true, and add comment about this is place the org.apache.fluss.flink.source.FlussSourceBuilder#setBounded and compatility. And In org.apache.fluss.flink.source.FlussSourceBuilder#setBounded, set stoppingOffsetsInitializer as OffsetsInitializer.latest().
There was a problem hiding this comment.
Thanks. I kept bounded as the legacy batch-mode flag because it is used as streaming = !bounded. Setting it in the parameterized overload would incorrectly turn bounded streaming into batch execution.
The no-argument setBounded() now sets bounded = true, Boundedness.BOUNDED, and OffsetsInitializer.latest(), with a compatibility comment added.
| public static OffsetsInitializer toStoppingOffsetsInitializer(BoundedOptions boundedOptions) { | ||
| switch (boundedOptions.boundedMode) { | ||
| case UNBOUNDED: | ||
| return null; |
There was a problem hiding this comment.
Not return null, please reference kafka:
In DynamicKafkaSourceBuilder, if boundedOptions.boundedMode is UNBOUNDED, will use the default NoStoppingOffsetsInitializer:
DynamicKafkaSourceBuilder() {
this.kafkaStreamSubscriber = null;
this.kafkaMetadataService = null;
this.deserializationSchema = null;
this.startingOffsetsInitializer = OffsetsInitializer.earliest();
this.stoppingOffsetsInitializer = new NoStoppingOffsetsInitializer();
this.boundedness = Boundedness.CONTINUOUS_UNBOUNDED;
this.props = new Properties();
}There was a problem hiding this comment.
Updated. Unbounded reads now use NoStoppingOffsetsInitializer, so the stopping initializer is always non-null. Batch reads still default to OffsetsInitializer.latest().
| * user-supplied stopping offsets. A bounded read only performs a one-time partition discovery, | ||
| * since partitions created after startup are outside the bounded range captured at startup. | ||
| */ | ||
| private final boolean bounded; |
There was a problem hiding this comment.
Maybe we can use org.apache.flink.api.connector.source.Boundedness as what kafka connector does.
There was a problem hiding this comment.
Updated. Boundedness is now propagated explicitly from the builder or table source to FlinkSource and FlinkSourceEnumerator.
| streaming ? new NoStoppingOffsetsInitializer() : OffsetsInitializer.latest(); | ||
| stoppingOffsetsInitializer != null | ||
| ? stoppingOffsetsInitializer | ||
| : (streaming |
There was a problem hiding this comment.
This can done in FlussSourceBuilder.
There was a problem hiding this comment.
Updated. The stopping initializer is now resolved before constructing the enumerator, and the enumerator requires a non-null value.
| this.splitPerAssignmentBatchSize = splitPerAssignmentBatchSize; | ||
| this.initialDiscoveryFinished = initialDiscoveryFinished; | ||
| this.unassignedSplits = new ArrayList<>(unassignedSplits); | ||
| this.noMoreNewSplits = initialDiscoveryFinished && isBoundedStreamingRead(); |
There was a problem hiding this comment.
This is incorrent, we can see what kafka says:
// This flag will be marked as true if periodically partition discovery is disabled AND the
// initializing partition discovery has finished.
private boolean noMoreNewPartitionSplits = false;There was a problem hiding this comment.
Updated. I removed the constructor initialization based on restored state. noMoreNewSplits is now set only after initialization finishes when periodic partition discovery is disabled.
| } | ||
|
|
||
| if (scanPartitionDiscoveryIntervalMs > 0) { | ||
| if (restoredBoundedPartitionSet) { |
There was a problem hiding this comment.
No need to make the thread more complicate, when restart read new partition is also no pronlem.
There was a problem hiding this comment.
Updated. I removed the bounded-restore special case. A restored bounded source now follows the normal one-time partition discovery path.
| return; | ||
| } | ||
| if (t != null) { | ||
| if (isBoundedStreamingRead()) { |
There was a problem hiding this comment.
Updated. I removed the bounded-specific partition discovery failure handling and kept the existing generic behavior.
|
Thanks @loserwang1024 for the detailed review! I’ve addressed the comments and aligned the implementation more closely with Kafka’s model. Please take another look when convenient. |
loserwang1024
left a comment
loserwang1024
left a comment
There was a problem hiding this comment.
totally agree with you, only some minor advice
| PushdownUtils.computeAvailableStatsColumns(flussRowType, tableConfig); | ||
| } | ||
|
|
||
| public FlinkTableSource( |
There was a problem hiding this comment.
Is this overload kept only for backward compatibility? FlinkTableSource is not a documented public API, and DataStream users access the source through FlussSourceBuilder. Do we need to preserve this constructor? Otherwise, could the existing constructor delegate directly to the new one with BoundedOptions.unbounded()?
| // This flag preserves the execution semantics of the legacy no-argument setBounded(), which | ||
| // switches the source to batch mode. The parameterized setBounded(OffsetsInitializer) keeps | ||
| // streaming execution semantics and only makes the source bounded. | ||
| private boolean bounded; |
There was a problem hiding this comment.
Could we remove this bounded flag? The legacy setBounded() conflates source boundedness with Flink's batch execution mode. Now that boundedness is represented explicitly by Boundedness, we should stop deriving the streaming argument from this flag. Batch-specific behavior can be determined from the actual execution mode in FlussSource in a follow-up. Also, please deprecate the no-argument setBounded() with both @deprecated and an @deprecated Javadoc tag.
2 participants
Purpose
Linked issue: close #4021
Brief change log
Tests
API and Format
Documentation